package org.pengge.rxjava.web.controller.first;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;
import org.pengge.rxjava.web.api.first.ChapterTwoApi;
import org.pengge.rxjava.web.entity.vo.ResponseVO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @author
 * @since 2017/7/21 下午4:07
 * Purpose
 */
@Slf4j
@RequestMapping("/first/chapter2")
@Controller
public class ChapterTwoController {
    @Autowired
    private ChapterTwoApi chapterTwoApi;

    @RequestMapping("/case1")
    @ResponseBody
    public ResponseVO case1() {
        ResponseVO responseVO = new ResponseVO();
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                log.info("Observable thread is : " + Thread.currentThread().getName());
                log.info("emit 1");
                emitter.onNext("Observable thread is : " + Thread.currentThread().getName());
                emitter.onComplete();
            }
        });

        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String integer) throws Exception {
                log.info("Observer thread is :" + Thread.currentThread().getName());
                log.info("onNext: " + integer);
            }
        };

        observable.subscribe(consumer);
        responseVO.setData(observable.blockingIterable());
        return responseVO;
    }

    @RequestMapping("/case2")
    @ResponseBody
    public ResponseVO case2() {
        ResponseVO responseVO = new ResponseVO();
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                log.info("Observable thread is : " + Thread.currentThread().getName());
                log.info("emit 1");
                emitter.onNext("Observable thread is : " + Thread.currentThread().getName());
                emitter.onComplete();
            }
        });

        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String integer) throws Exception {
                log.info("Observer thread is :" + Thread.currentThread().getName());
                log.info("onNext: " + integer);
            }
        };

        observable.subscribeOn(Schedulers.io()).observeOn(Schedulers.io()).subscribe(consumer);
        responseVO.setData(observable.subscribeOn(Schedulers.io()).blockingIterable());
        return responseVO;
    }

    /**
     * 实现了一个怎么使用异步线程来
     *
     * @return
     */
    @RequestMapping("/case3")
    @ResponseBody
    public ResponseVO case3() {
        ResponseVO responseVO = new ResponseVO();

        Date date = new Date();
        List<String> results = new ArrayList<>();
        Observable<String> stringObservable = chapterTwoApi.queryBaidu().observeOn(Schedulers.io());
        stringObservable.subscribeOn(Schedulers.io()).subscribe(results::add);
        System.out.println(new Date().getTime() - date.getTime());
        System.out.println("主线程开始休眠");
        try {
            Thread.sleep(10 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(new Date().getTime() - date.getTime());
        responseVO.setData(results);
        return responseVO;
    }
}
